package com.yanxu.consumer;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2024/5/23-2024
 * @Description: 这里用一句话描述这个类的作用
 */
public class PullConsumer {
    public static void main(String[] args) throws Exception {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer();
        consumer.setNamesrvAddr("139.129.91.21:9876");

        //获取所有的对列。
        Collection<MessageQueue> messageQueues = consumer.fetchMessageQueues("test");
        List<MessageQueue> messageQueueList = messageQueues.stream().filter(Objects::nonNull).collect(Collectors.toList());
        //assign 分配需要消费的队列
        consumer.assign(messageQueueList);
        //指定具体某一个队列从什么位置消费
        consumer.seek(messageQueueList.get(0),5);
        //poll 主动拉取消息
        List<MessageExt> messageExts = consumer.poll();
        for (MessageExt messageExt : messageExts) {
            System.out.println(messageExt);
        }

        consumer.start();

    }
}